package com.yunhe.transfer.service.base.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.transfer.client.MeasHistoryClient;
import com.yunhe.transfer.client.MeasRealtimeClient;
import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.DataTransferConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.common.model.base.DataTransferDTO;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.transfer.service.base.DataTransferService;
import com.yunhe.transfer.util.MeasurementUtil;
import com.yunhe.common.util.RedisClient;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.UUID;

/**
 * @title: ReceiveDataServiceImpl
 * @Author Chen Lun
 * @Date: 2021/1/20
 */
@Service
public class DataTransferServiceImpl implements DataTransferService {
    @Autowired
    private RedisClient redisClient;
    @Autowired
    private AuthorityClient authorityClient;
    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private MeasRealtimeClient measRealtimeClient;

    private static final Logger logger = LoggerFactory.getLogger(DataTransferServiceImpl.class);

    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Async
    @Override
    public void receiveData(List<Measurement> measurements, Long stationCode) throws JsonProcessingException {
//        Map<Long, Double> pvMap = new HashMap<>();
        // 根据电站获得时区（为了提升性能，使用缓存）
//        String timezone = getTimeZone(stationCode);
        if (CollectionUtil.isNotEmpty(measurements)) {
//            MeasurementUtil.processMeasurements(measurements, timezone);
//            // 保存到实时量测
//            for (Measurement item : measurements) {
//                //ObjectSave
//                redisClient.setObject(CacheConsts.KEY.MEASURE_OBJ.value(item.getPointNumber()), item);
//                pvMap.put(item.getPointNumber(), item.getVal());
//            }
//            // 发送实时值通知Kafka消息 - 目前只针对于实时告警
//            kafkaTemplate.send(KafkaConsts.Topic.Realtime, MAPPER.writeValueAsString(pvMap));
            //保存实时值（此处不发Kafaka消息，改为实时量测服务统一发送）
            DomainList<Measurement> domainList = new DomainList<>(measurements);
            measRealtimeClient.saveObj(domainList);
            // 保存到历史量测
            measHistoryClient.save(domainList, MeasurementConsts.FREQUENCY.Original.value());
        }
    }

    /**
     * 云端主动补招
     * @param dataTransferDTO
     * @throws JsonProcessingException
     */
    @Async
    @Override
    public void recollectData(DataTransferDTO dataTransferDTO) throws JsonProcessingException, ArgumentErrorException {
        List<Measurement> measurements = dataTransferDTO.getMeasurements();
        Long stationCode = dataTransferDTO.getStationCode();
        if (stationCode == null) {
            throw new ArgumentErrorException("电站代号不能为空");
        }
        String timezone = getTimeZone(stationCode);
        if (CollectionUtil.isNotEmpty(measurements)) {
            saveHistory(MeasurementUtil.processMeasurements(measurements, timezone));
            if (dataTransferDTO.getStartRtime() == null || dataTransferDTO.getEndRtime() == null) {
                dataTransferDTO = MeasurementUtil.recollectData(measurements, stationCode);
            }
            //消息太大无法发送，改用缓存进行处理
            String key = CacheConsts.KEY.RECOLLECT.value(UUID.randomUUID());
            redisClient.setObject(key, dataTransferDTO, CacheConsts.getDefaultTime());
            kafkaTemplate.send(KafkaConsts.Topic.Recollect, key);
        }
    }

    /**
     * MQTT主动补招
     * @param stationId
     * @param dtime
     * @throws JsonProcessingException
     */
    @Override
    public void recollectDataForMqtt(Long stationId, List<String> dtime) throws JsonProcessingException, ArgumentErrorException {
        DataTransferDTO dataTransferDTO = new DataTransferDTO();
        Long stationCode = getStationCode(stationId);
        if (stationCode == null) {
            throw new ArgumentErrorException("电站代号为空");
        }
        dataTransferDTO.setStationCode(stationCode);
        dataTransferDTO.setDtime(dtime);
        kafkaTemplate.send(KafkaConsts.Topic.RecollectOrder, MAPPER.writeValueAsString(dataTransferDTO));
    }

    /**
     * WebSocket主动补招
     * @param stationId
     * @param dtime
     * @throws JsonProcessingException
     */
    @Override
    public void recollectDataForWebSocket(Long stationId, List<String> dtime) throws JsonProcessingException, ArgumentErrorException {
        DataTransferDTO dataTransferDTO = new DataTransferDTO();
        Long stationCode = getStationCode(stationId);
        if (stationCode == null) {
            throw new ArgumentErrorException("电站代号为空");
        }
        dataTransferDTO.setStationCode(stationCode);
        dataTransferDTO.setDtime(dtime);
        dataTransferDTO.setType(DataTransferConsts.TransferType.RECOLLECT.value());
        kafkaTemplate.send(KafkaConsts.Topic.RecollectOrder, MAPPER.writeValueAsString(dataTransferDTO));
    }

    /**
     * 根据项目代号获得电站时区
     * @param stationCode
     * @return
     */
    private String getTimeZone(Long stationCode) {
        String timeZone = null;
        try {
            Station station = authorityClient.getStationByCode(stationCode).getBody().getResults();
            if (station != null) {
                timeZone = station.getTimeZone();
            }
            station = authorityClient.getStationById(stationCode).getResults();
            if (station != null) {
                timeZone = station.getTimeZone();
            }
        } catch (Exception e) {
            logger.error("获取电站时区报错", e);
        }
        return timeZone;
    }

    /**
     * 保存历史量测
     * @param measurements
     */
    public void saveHistory(List<Measurement> measurements) {
        DomainList<Measurement> domainList = new DomainList<>(measurements);
        measHistoryClient.save(domainList, MeasurementConsts.FREQUENCY.Original.value());
    }

    /**
     * 根据电站id获得电站code
     * @param stationId
     * @return
     */
    private Long getStationCode(Long stationId) {
        Long stationCode = null;
        try {
            Station station = authorityClient.getStationById(stationId).getResults();
            if (station != null) {
                stationCode = station.getCode();
            }
        } catch (Exception e) {
            logger.error("获取电站时区报错", e);
        }
        return stationCode;
    }
}
